/*
** Copyright (C) 2009 Joshua Brent Marsh <joshua@icub3d.com>
** 
** This file is part of icub3d-adt
**
** icub3d-adt is free software; you can redistribute it and/or modify
** it under the terms of the GNU General Public License as published by
** the Free Software Foundation; either version 3 of the License, or
** (at your option) any later version.
**  
** This program is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
** GNU General Public License for more details.
**  
** You should have received a copy of the GNU General Public License
** along with this program; if not, write to the Free Software
** Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
**  
*/

/* $Id: thread_pool.c 41 2009-11-13 02:28:44Z joshua@icub3d.com $ */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include "thread_pool.h"

typedef
struct work_pair
{
  work_func *func;
  void *param;
} WorkPair;

WorkPair *
work_pair_create (work_func *func, void *param)
{
  WorkPair *wp;
  wp = malloc (sizeof (WorkPair));
  if (wp == NULL)
    return NULL;

  wp->func = func;
  wp->param = param;

  return wp;
}

void
work_pair_destroy (void *a)
{
  free (a);
}

void *
work_processor (void *in)
{
  ThreadPool *tp = in;
  WorkPair *wp;

  while (1)
    {
      pthread_mutex_lock (&(tp->queue_lock));
      
      /* If there are no items in the queue, and we are stopping, break. */
      if (pqueue_size (tp->work_queue) == 0 && tp->stopping)
	{
	  pthread_mutex_unlock (&(tp->queue_lock));
	  break;
	}
      /* If there are no items and we aren't stopping, wait for availability. */
      else if (pqueue_size (tp->work_queue) == 0)
	pthread_cond_wait (&(tp->empty_queue), &(tp->queue_lock));

      /* If we are set to stop, quit. */
      if (tp->stopped == 1)
	{
	  pthread_mutex_unlock (&(tp->queue_lock));
	  break;
	}

      /* Get the next item in the queue */
      wp = pqueue_dequeue (tp->work_queue);
      pthread_mutex_unlock (&(tp->queue_lock));

      if (wp == NULL)
	continue;

      /* Run the process. */
      wp->func (wp->param);
      work_pair_destroy (wp);
      wp = NULL;
    }

  pthread_exit (NULL);
}

ThreadPool * 
thread_pool_create (size_t num_threads)
{
  ThreadPool *tp;
  tp = malloc (sizeof (ThreadPool));
  if (tp == NULL)
    return NULL;

  tp->threads = malloc (sizeof (pthread_t) * num_threads);
  if (tp->threads == NULL)
    {
      free (tp);
      return NULL;
    }

  tp->work_queue = pqueue_create (MIN, work_pair_destroy);
  if (tp->work_queue == NULL)
    {
      free (tp->threads);
      free (tp);
      return NULL;
    }

  tp->num_threads = num_threads;
  tp->stopped = 1;
  tp->stopping = 1;

  return tp;
}

void
thread_pool_start (ThreadPool *tp)
{
  pthread_attr_t attr;
  size_t x;

  if (tp->stopped == 0)
    return;

  pthread_mutex_init (&(tp->queue_lock), NULL);
  pthread_cond_init (&(tp->empty_queue), NULL);
  
  pthread_attr_init (&attr);
  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_JOINABLE);

  for (x = 0; x < tp->num_threads; x++)
    pthread_create (&(tp->threads[x]), &attr, work_processor, (void *)tp);

  tp->stopped = 0;
  tp->stopping = 0;
}

void
thread_pool_stop (ThreadPool *tp, int remaining)
{
  size_t x;

  if (tp->stopped == 1)
    return;

  pthread_mutex_lock (&(tp->queue_lock));
  
  /* Set it to stop based on remaining. */
  tp->stopping = 1;

  if (!remaining)
    tp->stopped = 1;
  
  /* Send out the signal to all the processors, just in case they are in a 
     waiting state. */
  pthread_cond_broadcast (&(tp->empty_queue));
  pthread_mutex_unlock (&(tp->queue_lock));
  
  for (x = 0; x < tp->num_threads; x++)
    pthread_join (tp->threads[x], NULL);

  /* Clean up these. */
  pthread_mutex_destroy (&(tp->queue_lock));
  pthread_cond_destroy (&(tp->empty_queue));

  tp->stopped = 1;
  tp->stopping = 1;
}

void
thread_pool_destroy (ThreadPool *tp, int remaining)
{
  thread_pool_stop (tp, remaining);
  pqueue_destroy (tp->work_queue);
  free (tp->threads);
  free (tp);
}

int
thread_pool_enqueue (ThreadPool *tp, size_t priority, 
		     work_func *wfunc, void *param)
{
  WorkPair *wp;

  if (tp->stopping == 1)
    return 0;

  wp = work_pair_create (wfunc, param);
  if (wp == NULL)
    return 0;

  pthread_mutex_lock (&(tp->queue_lock));

  if (pqueue_enqueue (tp->work_queue, priority, wp) != wp)
    {
      pthread_mutex_unlock (&(tp->queue_lock));
      work_pair_destroy (wp);
      return 0;
    }

  if (pqueue_size (tp->work_queue) == 1)
    pthread_cond_broadcast (&(tp->empty_queue));

  pthread_mutex_unlock (&(tp->queue_lock));

  return 1;
}

